
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include "ioloop.h"

namespace datashuttle
{

EventLoop::EventLoop()
	:mEventHandle(0),mbRunning(false)
{
}

EventLoop::~EventLoop()
{
	clearResource();
	mBreaker = 0;
}

bool EventLoop::init( int eventHandle , BreakerPtr breaker )
{
	clearResource();
	
	if( breaker != 0 )
	{
		mBreaker = breaker;
	}
	else
	{
		mBreaker = new Breaker(*this);
	}

	if( !mBreaker->init())
		return false;

	mEventHandle = eventHandle;

	if(!mBreaker->addToIoLoop())
		return false;
	//add breaker reader side into epoll
	mbRunning = true;
	return true;	
}

bool EventLoop::addFd( EventTaskPtr task )
{
	if(! task ) return false;

	int op = EPOLL_CTL_ADD;

	struct epoll_event evt;

	//采用EdgeTrigger的方式
	evt.events = EPOLLIN | EPOLLOUT | EPOLLERR | EPOLLHUP | EPOLLET; 
	evt.data.ptr = task.get();

	//TODO: 当前的设计采用了智能指针的方式来管理一个instance的生命周期
	//      当前的问题是由于epoll需要的是void*指针，那么当epoll还有这个instance的指针的时候
	//      可能由于其他部分已经不还含有这个instance智能指针造成instance被delete。最终可能
	//      会导致程序crash
	int r = epoll_ctl(mEventHandle , task->getFd() ,op , &evt );
	if( 0 == r )
	{//success
		return true;
	}
	else
	{//fail
		//TODO: log some thing here
		return false;
	}
}

void EventLoop::removeFd( EventTaskPtr task)
{
	if(! task ) return;
	struct epoll_event evt;
	memset(&evt , 0, sizeof(evt) );
	int r =  epoll_ctl(mEventHandle , EPOLL_CTL_DEL , task->getFd() , &evt );
	if( 0 != r )
	{//TODO: log something here
	}
}

bool EventLoop::addTask( BaseTaskPtr task )
{
	if(!task) return false;
	size_t colId = (size_t)-1;
	size_t rowId = mTimerTable.addTask( task , task->timeToExpire() , &colId );
	task->setTimerTablePosition(rowId, colId);
	return true;
}

void EventLoop::removeTask( BaseTaskPtr task )
{
	if(! task ) return ;
	mTimerTable.removeTask( task->timerTableRowId(), task->timerTableColId() );	
}

void EventLoop::stop()
{
	mbRunning = false;
	mBreaker->wakeup();
}

long long EventLoop::runExpiredTasks()
{
	long long now = timenow();
	long long delta = now - mTimerTable.targetTime();
	if( delta > 0 )
		return delta;

	BaseTaskPtr task = NULL;
	long long tte = 0;
	do
	{
		task = mTimerTable.getTask( tte );
		
		task->run(TYPE_TASK_TIMER);

	}while( task != 0 );
	now = timenow();
	delta = now - mTimerTable.targetTime();
	if( delta < 0 )
		delta = 0 ;

	return delta;
}

void EventLoop::postRun()
{
	if( mBreaker )
	{
		mBreaker->removeFromIoLoop();
	}
}

void EventLoop::run()
{
	struct epoll_event evts[64];
	while( mbRunning )
	{
		int tte = static_cast<int>(runExpiredTasks());

		int r = epoll_wait( mEventHandle , evts , 64 , tte );

		if( !mbRunning ) 
			break;

		if( r < 0 )
		{//timed out
			// log error message here
			// 需要在这里停止epoll的运行吗?
		}
		else if( r > 0 )
		{
			for( int i = 0 ; i < r ; i ++ )
			{
				EventTaskPtr task = reinterpret_cast<EventTask*>( evts[i].data.ptr );
				if( ! task )
					continue;

				int type = TYPE_TASK_NULL;				
				if( evts[i].events & EPOLLIN )
					type |= TYPE_TASK_DATAIN;
				if( evts[i].events & EPOLLOUT )
					type |= TYPE_TASK_DATAOUT;
				if( evts[i].events & EPOLLERR)
					type |= TYPE_TASK_ERROR;

				//TODO: 在真实的运行环境里面是否需要catch exception
				task->run(type );
			}
		}
	}
	postRun();
}

/////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////

Breaker::Breaker(IoLoop& loop )
	:EventTask(loop)
{
	memset(mBreakerHandle,0,sizeof(mBreakerHandle));
}
Breaker::~Breaker()
{
	clearResource();
}

void Breaker::clearResource()
{
	if( mBreakerHandle[0] > 0 )
		close( mBreakerHandle[0] );
	if( mBreakerHandle[1] > 0 )
		close( mBreakerHandle[1] );
	memset(mBreakerHandle,0,sizeof(mBreakerHandle));
}

bool Breaker::init()
{
	if( !createBreaker() )
		return false;
	
	updateFd( mBreakerHandle[0] );

	return true;
}

bool Breaker::createBreaker()
{
	clearResource();
	
	if( pipe( mBreakerHandle ) < 0 ) //create pipe
		return false;
	
	if( fcntl( mBreakerHandle[0], F_SETFL , O_NONBLOCK ) )
	{
		close( mBreakerHandle[0] );
		close( mBreakerHandle[1] );
		return false;
	}
	return true;
}
void Breaker::clearInterrupt()
{
	do
	{
		char t[8];
		ssize_t c = read(mBreakerHandle[0],	t, sizeof(t) );
		if( c < 0 )
		{
			if( errno == EAGAIN )
				break;// no more data
			//we should rebuild the breaker because something wrong happened to it
			createBreaker();
			break;
		}
	}while(true);
}

void Breaker::setInterrupt()
{
	char c = '&';
	
	write(mBreakerHandle[1],&c,1); // how to deal with th failure?

	//TODO: I should check the return value
	//      If there is any error occurred, the breaker should be re-built
	//      But this is a difficult thing in multi-thread environment
}
void Breaker::wakeup()
{
	setInterrupt();
}

void Breaker::run( int type )
{
	clearInterrupt();
}

}//namespace datashuttle



